package com.atguigu.sql;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.call;

/**
 * @author gmd
 * @desc 自定义TableFunction。使用Table API进行流数据的处理
 * @since 2024-12-04 07:15:06
 */
public class MyTableFunction {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> strDS = env.fromElements(
                "hello flink",
                "ni hao java",
                "kong ni qi wa hadoop"
        );

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 创建视图，wordsTable
        Table sensorTable = tableEnv.fromDataStream(strDS, $("words"));
        tableEnv.createTemporaryView("wordsTable", sensorTable);

        // 注册函数
        tableEnv.createTemporaryFunction("MySplitFunction", MySplitFunction.class);

        // 调用自定义函数
        tableEnv
                // 交叉联结
                // .sqlQuery("select words,word,length from wordsTable,lateral table(MySplitFunction(words))")
                // 带on true 条件的左联结
                // .sqlQuery("select words,word,length from wordsTable left join lateral table(MySplitFunction(words)) on true")
                // 重命名侧向表中的字段
                .sqlQuery("select words, newWord, newLength from wordsTable left join lateral table(MySplitFunction(words)) as T(newWord, newLength) on true")
                .execute()
                .print();
    }

    // 注解可以指定输入和输出的数据类型。
    // 类型标注：Row，包含两个字段：word和length。继承TableFunction<返回的类型>
    @FunctionHint(output = @DataTypeHint("ROW<word STRING, length INT>"))
    public static class MySplitFunction extends TableFunction<Row> {
        // 返回是void，用collect方法输出
        public void eval(String str) {
            for (String word : str.split(" ")) {
                collect(Row.of(word, word.length()));
            }
        }
    }

    /*
     * select words, newWord, newLength from wordsTable left join lateral table(MySplitFunction(words)) as T(newWord, newLength) on true
     * 这条SQL语句使用了Flink SQL和自定义表函数来处理数据。具体而言，这条SQL语句的含义如下：
     *
     * 1. **`select words,newWord,newLength`**: 选择最终结果集的三个字段，分别是`words`、`newWord`和`newLength`。
     * 2. **`from wordsTable`**: 从名为`wordsTable`的表中读取数据。
     * 3. **`left join lateral table(MySplitFunction(words)) as T(newWord,newLength)`**:
     *     - **`left join`**: 进行左外连接，即使右侧表（由`MySplitFunction`生成的表）没有匹配的行，也会保留左侧表（`wordsTable`）的所有行。
     *     - **`lateral`**: 表示侧向连接，即对每一行执行表函数。
     *     - **`table(MySplitFunction(words))`**: 调用自定义的表函数`MySplitFunction`，传入`words`列的值。`MySplitFunction`会将`words`列的字符串拆分成多个单词和它们的长度，并返回一个包含这些数据的表。
     *     - **`as T(newWord,newLength)`**: 给由表函数生成的表起一个别名`T`，并将表中的字段重命名为`newWord`和`newLength`。
     * 4. **`on true`**: 连接条件总是为真，确保左外连接的每一行都能匹配到右侧表中的所有行。
     *
     * 总结下来，这条SQL语句的作用是：
     * - 从`wordsTable`中读取数据。
     * - 对每一行的`words`列使用自定义表函数`MySplitFunction`进行拆分，生成多个新行，每个新行包含一个单词及其长度。
     * - 将生成的新行与原始行进行左外连接，保留原始行的所有数据，并添加拆分后的单词及其长度。
     */

}
